package com.pactera.asmp.server.common.websocket;

import com.pactera.asmp.server.common.lock.Lock;
import com.pactera.asmp.server.common.logprint.utils.CollectionUtils;
import com.pactera.asmp.server.job.dispatch.impl.feedback.FeedbackManager;
import com.pactera.asmp.server.pojo.SocketMessage;
import com.pactera.asmp.server.service.IDeviceServcie;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
@EnableScheduling
public class HeartBeatService {

    public static final String REDIS_KEY_WS_ONLINE_MEMBERS = "WS_ONLINE_MEMBERS";
    public static final String REDIS_KEY_WS_ONLINE_DUID_Simulator= "WS_ONLINE_DUID_Simulator";

    @Autowired
    private StringRedisTemplate redisService;

    @Autowired
    @Lazy
    private SocketUtil socketUtil;

    @Autowired
    private IDeviceServcie deviceServcie;

    @Autowired
    private Lock lock;

    private volatile Boolean active = true;

    @Value("${heartBeat.clear.schedule.enabled}")
    private Boolean scheduleEnable;

    @Autowired
    @Lazy
    private FeedbackManager feedbackManager;

    // 用户心跳上报续期
    /**
     * 添加用户在线信息
     * @param memberId
     * @return
     */
    @Async
    public Boolean online(String memberId) {
        if(Boolean.FALSE.equals(active)) return true;
        if(StringUtils.isEmpty(memberId) || "null".equalsIgnoreCase(memberId)) return false;
        return this.redisService.opsForZSet().add(REDIS_KEY_WS_ONLINE_MEMBERS, memberId, Instant.now().toEpochMilli());
    }


    /**
     * 移除用户
     */
    @Async
    public void remove(String... memberId){
        if(Boolean.FALSE.equals(active)) return;

        this.redisService.opsForZSet().remove(REDIS_KEY_WS_ONLINE_MEMBERS,memberId);

        feedbackManager.feedbackForDeviceOffline(memberId[0].split("@")[0]);
    }

    /**
     * 清除超过一定时间没在线的用户数据
     * @param duration
     * @return
     */
    public void clear(Duration duration) {
        if(Boolean.FALSE.equals(active)) return;

        long maxScore = LocalDateTime.now().minus(duration).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

        Set<String> members = this.redisService.opsForZSet().rangeByScore(REDIS_KEY_WS_ONLINE_MEMBERS, 0, maxScore);
        // 下线这些用户的ws session
        if(CollectionUtils.isEmpty(members)) return ;

        Set<String> duids = members.stream().map(s -> s.split("@")[0]).filter(s->s.length() == 36).collect(Collectors.toSet());
        if(!CollectionUtils.isEmpty(duids)) {
            this.deviceServcie.deviceOfflineByDuids(duids);
            //通知
            for(String duid: duids){
                feedbackManager.feedbackForDeviceOffline(duid);
            }
        }

        String prefixId ;
        String otherId;

        for (String member : members){
            prefixId = member.split("@")[0];
            otherId = redisService.opsForValue().get(prefixId);
            this.clearHandler(member,prefixId,otherId);
        }

        this.redisService.opsForZSet().removeRangeByScore(REDIS_KEY_WS_ONLINE_MEMBERS, 0,maxScore);
    }

    private void clearHandler(String member,String prefixId,String otherId) {
        String serverId;
        if(prefixId.startsWith("screen:")){
            serverId = (String) redisService.opsForHash().get(SocketUtil.REDIS_KEY_SOCKET_SERVER_SCREEN,member);
            redisService.opsForSet().remove(SocketUtil.REDIS_KEY_WS_SCREEN_INSTANCE+":"+prefixId,member);
            redisService.opsForHash().delete(SocketUtil.REDIS_KEY_SOCKET_SERVER_SCREEN,member);

            //清除项目绑定
            Object objPro = redisService.opsForHash().get(SocketUtil.REDIS_KEY_HASH_SCREEN_SESSION_PROJECT, member);
            if(objPro != null){
                String projectId = objPro.toString();
                redisService.opsForSet().remove(SocketUtil.REDIS_KEY_SET_SCREEN_PROJECT_SESSION+":"+projectId,member);
            }
            redisService.opsForHash().delete(SocketUtil.REDIS_KEY_HASH_SCREEN_SESSION_PROJECT, member);

            //清除设备绑定(image显示设备)
            Set<String> ssd = redisService.opsForSet().members(SocketUtil.REDIS_KEY_SET_SCREEN_SESSION_DEVICE + ":" + member);
            if(!CollectionUtils.isEmpty(ssd)){
                for (String duid : ssd){
                    redisService.opsForSet().remove(SocketUtil.REDIS_KEY_SET_SCREEN_DEVICE_SESSION+":"+duid,member);
                }
            }
            redisService.delete(SocketUtil.REDIS_KEY_SET_SCREEN_SESSION_DEVICE+":"+member);

            //清除上位机绑定
            Set<String> membersComputer = redisService.opsForSet().members(SocketUtil.REDIS_KEY_SET_SCREEN_SESSION_COMPUTER + ":" + member);
            if(!CollectionUtils.isEmpty(membersComputer)){
                for (String cuid : membersComputer){
                    redisService.opsForSet().remove(SocketUtil.REDIS_KEY_SET_SCREEN_COMPUTER_SESSION+":"+cuid,member);
                }
            }
            redisService.delete(SocketUtil.REDIS_KEY_SET_SCREEN_SESSION_COMPUTER + ":" + member);

            //清除设备绑定
            Set<String> membersDuid = redisService.opsForSet().members(SocketUtil.REDIS_KEY_SET_SCREEN_SESSION_DUID + ":" + member);
            if(!CollectionUtils.isEmpty(membersDuid)){
                for (String duid : membersDuid){
                    redisService.opsForSet().remove(SocketUtil.REDIS_KEY_SET_SCREEN_DUID_SESSION+":"+duid,member);
                }
            }
            redisService.delete(SocketUtil.REDIS_KEY_SET_SCREEN_SESSION_DUID + ":" + member);
        }
        else if (prefixId.length() == 36){
            serverId = (String) redisService.opsForHash().get(SocketUtil.REDIS_KEY_SOCKET_SERVER_DEVICE,member);
            redisService.opsForSet().remove(SocketUtil.REDIS_KEY_WS_DEVICE_INSTANCE+":"+prefixId,member);
            redisService.opsForHash().delete(SocketUtil.REDIS_KEY_SOCKET_SERVER_DEVICE,member);
        }
        else {
            serverId = (String) redisService.opsForHash().get(SocketUtil.REDIS_KEY_SOCKET_SERVER_USER,member);
            redisService.opsForSet().remove(SocketUtil.REDIS_KEY_WS_USER_INSTANCE+":"+prefixId,member);
            redisService.opsForHash().delete(SocketUtil.REDIS_KEY_SOCKET_SERVER_USER,member);
        }
        if(!StringUtils.isEmpty(prefixId) && prefixId.startsWith("CAN")){
            redisService.delete(prefixId.substring(3));
        }
        else {
            redisService.delete(prefixId);
        }
        if(!StringUtils.isEmpty(otherId)) redisService.delete(otherId);
        log.info("HeartBeat Timeout----------------{},--------------routingKey:{}",member,serverId);
        if(!StringUtils.isEmpty(serverId)) {
            socketUtil.sendMQ(serverId,new SocketMessage("Server",member,"timeout",null,null));
        }
        redisService.opsForZSet().remove(REDIS_KEY_WS_ONLINE_MEMBERS,member);

    }

    @PostConstruct
    public void init(){
        log.info("ZoneId.systemDefault() : {}",ZoneId.systemDefault());
        this.cleanExpireMember();
        this.cleanExpireDuidSimulator();
    }

    // 计划任务检测超时用户
    @Scheduled(cron = "0 0/2 * * * ? ")
    public void cleanExpireMember(){
        if(Boolean.FALSE.equals(scheduleEnable)) return;
        if(Boolean.FALSE.equals(active)) return;
        String guid = UUID.randomUUID().toString();//FIXME 每个机器时间是不一致的，可能锁不住
        try {
            // 分布式锁。 只允许一个线程处理
            lock.lock(guid);
            // 清理指定时间内无心跳续约的客户端
            this.clear(Duration.ofSeconds(32));
        }catch (Exception e){
            e.printStackTrace();
        }
        finally {
            try {
                lock.unlock(guid);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    @Async
    public Boolean onlineDuidSimulator(String duid,String simulatorId) {
        if(Boolean.FALSE.equals(active)) return true;
        if(StringUtils.isEmpty(duid) || StringUtils.isEmpty(simulatorId)) return false;
        String memberId = duid +"_"+simulatorId;
        return this.redisService.opsForZSet().add(REDIS_KEY_WS_ONLINE_DUID_Simulator, memberId, Instant.now().toEpochMilli());
    }

    // 计划任务检测超时设备
    @Scheduled(cron = "0 0/2 * * * ? ")
    public void cleanExpireDuidSimulator(){
        if(Boolean.FALSE.equals(scheduleEnable)) return;
        if(Boolean.FALSE.equals(active)) return;
        this.clearDuidSimulator(Duration.ofSeconds(32));
    }

    private void clearDuidSimulator(Duration duration) {
        if(Boolean.FALSE.equals(active)) return;
        long maxScore = LocalDateTime.now().minus(duration).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
        Set<String> members = this.redisService.opsForZSet().rangeByScore(REDIS_KEY_WS_ONLINE_DUID_Simulator, 0, maxScore);

        // 下线这些用户的ws session
        if(CollectionUtils.isEmpty(members)) return ;

        String duid;
        String simulatorId;
        int duidStatus;
        for(String id : members){
            duidStatus = 2;
            duid = id.split("_")[0];
            simulatorId = id.split("_")[1];
            Long rank = this.redisService.opsForZSet().rank(REDIS_KEY_WS_ONLINE_MEMBERS, duid);

            if(rank != null && rank.intValue() > -1){
                duidStatus =1;
            }
            this.deviceServcie.deviceOfflineBySimulatorId(duid,simulatorId,duidStatus);
            this.redisService.delete(duid);
            this.redisService.delete(simulatorId);
        }
        this.redisService.opsForZSet().removeRangeByScore(REDIS_KEY_WS_ONLINE_DUID_Simulator, 0,maxScore);
    }
}
